package org.orbis.communicate.rpc;

import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.SortedSet;
import java.util.concurrent.locks.ReentrantLock;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.BooleanType;
import org.msgpack.object.FloatType;
import org.msgpack.object.IntegerType;
import org.msgpack.object.NilType;
import org.msgpack.object.RawType;
import org.orbis.communicate.FlagsResult;
import org.orbis.communicate.MainService;
import org.orbis.communicate.deleter.DeleterPackage;
import org.orbis.communicate.filter.FieldFilter;
import org.orbis.communicate.indexer.content.ContentIndexerClient;
import org.orbis.communicate.indexer.term.TermIndexerClient;
import org.orbis.communicate.indexer.term.TermPackage;
import org.orbis.communicate.lock.IndexableSwitch;
import org.orbis.communicate.node.Node;
import org.orbis.index.IndexingPackage;
import org.orbis.logging.LogParser;
import org.orbis.memorymanager.TermInformation;
import org.orbis.search.SearchPackage;
import org.orbis.search.SearchResult;
import org.orbis.structure.LookupPointer;

public final class PeerHandler implements PeerHandlerInterface{
	private MainService mains;

	public PeerHandler(MainService peerService){
		this.mains = peerService;
	}

	/** Send a node chain. */
	@Override
	public ArrayType nodeList(){
		LogParser.finest("nodeList - called.");
		int count = 0;
		SortedSet<Node> currentInfos = mains.getNode();
		MessagePackObject[] output = new MessagePackObject[2 * currentInfos.size()];

		for(Node eachNode: currentInfos){
			output[count++] = RawType.create(eachNode.getAddr());
			output[count++] = IntegerType.create(eachNode.getPort());
		}
		return ArrayType.create(output);
	}

	/** Registers node to chains. */
	@Override
	public boolean register(String IPaddr, int port){
		LogParser.finest("register - called. " + IPaddr + ":" + port);
		try{
			mains.addNode(Node.create(IPaddr, port));
		}catch(UnknownHostException ignored){
			return false;
		}
		return true;
	}

	/** @param IPaddr
	 * @param port */
	@Override
	public void deleteNode(String IPaddr, int port){
		LogParser.finest("deleteNode - called.");
	}

	private static FieldFilter joinFilter;
	private static ReentrantLock transaction;
	private static long[] keys;
	private static int[] copyTotals;
	private static int[] copyLefts;
	private static final int TRANSACTIONS = 16;

	static{
		joinFilter = new FieldFilter();
		transaction = new ReentrantLock();
		keys = new long[TRANSACTIONS];
		for(int i = 0; i < TRANSACTIONS; i++){
			keys[i] = -1L;
		}
		copyTotals = new int[TRANSACTIONS];
		copyLefts = new int[TRANSACTIONS];
	}

	/** Send datas specified by IDs. */
	@Override
	public ArrayType getData(int clusterID, long key){
		ArrayType returnValue = ArrayType.create(new MessagePackObject[]{NilType.create()});
		int keyIndex = -1;
		for(int i = 0; i < TRANSACTIONS; i++){
			if(keys[i] == key){
				keyIndex = i;
				break;
			}
		}

		if(keyIndex != -1){
			if(copyLefts[keyIndex] < 0) return returnValue;
			int index = copyTotals[keyIndex] - copyLefts[keyIndex];
			if(index == 0) returnValue = mains.getClusterPointInfo(clusterID);
			else returnValue = mains.toArrayType(clusterID, index - 1);

			copyLefts[keyIndex]--;
		}

		return returnValue;
	}

	/** Prepares for copying whole blocks specified by cluster ID.
	 * when copying file, stop indexing and searching process. */
	@Override
	public int getDataSize(int clusterID, long key){
		LogParser.finest("getDataSize - called.");
		int amount = -1;
		try{
			transaction.lock();
			IndexableSwitch.lock();

			amount = mains.getClusterBlockAmount(clusterID);
			for(int i = 0; i < TRANSACTIONS; i++){
				if(keys[i] == -1L){
					keys[i] = key;
					copyTotals[i] = amount + 1;
					copyLefts[i] = amount + 1;
				}
			}
		}finally{
			transaction.unlock();
		}

		return amount;
	}

	/** Call to make ended sending data. */
	@Override
	public boolean endDataSend(long ID){
		LogParser.finest("endDataSend - called.");

		for(int i = 0; i < TRANSACTIONS; i++){
			if(keys[i] == ID){
				keys[i] = -1L;
				copyTotals[i] = -1;
				copyLefts[i] = -1;
				return true;
			}
		}

		return false;
	}

	/** Stop data transfer forcedly
	 * 
	 * @param req */
	@Override
	public void forceStop(int ID){
		LogParser.finest("forceStop - called.");

		try{
			IndexableSwitch.unlock();
		}catch(InterruptedException e){}
	}

	@Override
	public void deleteContent(ArrayType deletePkg, int cluster){
		LogParser.finest("deleteContentLong - called.");
		mains.invokeDelete(DeleterPackage.unpack(deletePkg), cluster);
	}

	@Override
	public String isLive(){
		LogParser.finest("isLive - called.");
		return "Live";
	}

	/** Add specified content to the index. */
	@Override
	public void indexContent(ArrayType content){
		LogParser.finest("indexContent - called.");

		ContentIndexerClient.addToQueue(IndexingPackage.fromPackageContent(content));
	}

	/** Add specified term to the index. */
	public static int terms = 0;

	@Override
	public void indexTerm(ArrayType pkgObject){
		LogParser.finest("IndexTerm - called.");
		TermIndexerClient.addToQueue(TermPackage.unpack(pkgObject));
	}

	/** Search term from own index. */
	@Override
	public ArrayType searchTerm(ArrayType packed, int cluster){
		LogParser.finest("searchTerm - called.");
		SearchPackage pkg = SearchPackage.unpack(packed);
		boolean[] contentType = pkg.searchTypesBool();
		TermInformation info = mains.invokeSearch(pkg.query(), contentType, cluster);
		if(info == null) return null;
		FlagsResult[][] pkgs = new FlagsResult[contentType.length][];
		for(int i = 0; i < contentType.length; i++){
			if(contentType[i] == false) continue;
			LookupPointer invPtr = info.getDictionary().getInvertIndex(i);
			invPtr.withCluster(cluster);
			pkgs[i] = mains.invokeInvertIndex(invPtr, pkg);
		}
		FlagsResult[] result = joinFilter.join(contentType, pkgs);
		Arrays.sort(result, new Comparator<FlagsResult>(){
			@Override
			public int compare(FlagsResult arg0, FlagsResult arg1){
				return arg0.ID() < arg1.ID()? 1: arg0.ID() > arg1.ID()? -1: 0;
			}
		});

		int objsize = 1;
		objsize += result.length;

		MessagePackObject[] objs = new MessagePackObject[100];
		objs[0] = IntegerType.create(result.length);
		int subtotal = 1;
		for(int i = 0; i < result.length; i++){
			objs[subtotal] = IntegerType.create(result[i].ID());
			objs[subtotal + 1] = IntegerType.create(result[i].newest());
			subtotal += 2;
			boolean[] bins = result[i].binaryFlags();
			int binLength = bins.length;
			for(int j = 0; j < binLength; j++)
				objs[subtotal + j] = BooleanType.create(bins[j]);
			subtotal += binLength;

			int[] ints = result[i].intFlags();
			int intLength = ints.length;
			for(int j = 0; j < intLength; j++)
				objs[subtotal + j] = IntegerType.create(ints[j]);
			subtotal += intLength;

			long[] longs = result[i].longFlags();
			int longLength = longs.length;
			for(int j = 0; j < longLength; j++)
				objs[subtotal + j] = IntegerType.create(longs[j]);
			subtotal += longLength;

			float[] floats = result[i].floatFlags();
			int floatLength = longs.length;
			for(int j = 0; j < floatLength; j++)
				objs[subtotal + j] = FloatType.create(floats[j]);
		}

		return ArrayType.create(objs);
	}

	// Send content from specified ID.
	@Override
	public MessagePackObject getContent(long ID, int cluster){
		LogParser.finest("getContent - called.");
		int i;
		SearchResult content = mains.lookupContent(ID, cluster);
		MessagePackObject[] retobjs = new MessagePackObject[content.contentSize() + content.appendixSize()];

		for(i = 0; i < content.contentSize(); i++)
			retobjs[i] = RawType.create(content.content(i));
		for(i = 0; i < content.appendixSize(); i++)
			retobjs[i + content.contentSize()] = RawType.create(content.appendix(i));

		return ArrayType.create(retobjs);
	}

	// Obtain node list.
	@Override
	public ArrayType getList(int ptrBlock, int ptrPoint, long IDfrom, int length){
		LogParser.finest("getList - called.");
		return null;
	}

	/** Send fundamental information to join the network (BlockNum, BlockSize, Replication, IndexMode) */
	@Override
	public ArrayType getInfo(){
		LogParser.finest("getInfo - called.");
		return mains.putInfo();
	}

	@Override
	public int getIndexQueue(){
		return mains.getQueueSize(0);
	}

	@Override
	public int getSearchQueue(){
		return mains.getQueueSize(1);
	}

	@Override
	public void stopIndexing(){
		IndexableSwitch.lock();
	}

	@Override
	public void resumeIndexing(){
		try{
			IndexableSwitch.unlock();
		}catch(InterruptedException e){}
	}

	@Override
	public void revertLatest(){
	}

	@Override
	public void backupLatest(){
	}

	@Override
	public void snapshot(){
	}

	@Override
	public void synchronizeClock(){
		mains.synchronizeClock();
	}

	@Override
	public void suspend(String IPaddr, int port, long suspendTime, String reason){
		mains.suspend(IPaddr, port, suspendTime);
	}

	@Override
	public void resume(String IPaddr, int port){
		mains.resume(IPaddr, port);
	}
}
